package com.walkud.rxjava.creating;

import com.walkud.rxjava.utils.PrintLogSubscriber;
import io.reactivex.*;

import java.util.Arrays;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;


/**
 * 创建被观察者操作符简单示例
 * Create — creating an Observable from scratch by calling observer methods programmatically
 * Defer — do not creating the Observable until the observer subscribes, and creating a fresh Observable for each observer
 * Empty/Never/Throw — creating Observables that have very precise and limited behavior
 * From — convert some other object or data structure into an Observable
 * Interval — creating an Observable that emits a sequence of integers spaced by a particular time interval
 * Just — convert an object or a set of objects into an Observable that emits that or those objects
 * Range — creating an Observable that emits a range of sequential integers
 * Repeat — creating an Observable that emits a particular item or sequence of items repeatedly
 * Start — creating an Observable that emits the return value of a function
 * Timer — creating an Observable that emits a single item after a given delay
 * <p>
 * blockingSubscribe(在当前线程订阅，阻塞)
 */
public class CreatingSimple {

    public static void main(String[] agrs) {
//        creating();
//        defer();
//        empty();
//        never();
//        error();
//        from();
//        interval();
//        just();
//        range();
//        intervalRange();
//        repeat();
//        start();
//        timer();
    }

    /**
     * 创建一个延迟发射的被观察者
     */
    private static void timer() {
        //延迟2秒发射数据
        Observable.timer(2, TimeUnit.SECONDS).blockingSubscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个发射数据前先发射一个数据
     */
    private static void start() {
        Observable.just("RxJava").startWith("Start").subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个重复发射的被观察者
     */
    private static void repeat() {
        //重复发射3次（包含初次）
        Observable.just("repeat").repeat(3).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个范围周期循环发射的被观察者
     */
    private static void intervalRange() {
        //从3开始发射，共发射10次，初始延迟0秒，之后期间延迟1秒
        Observable.intervalRange(5, 10, 0, 1, TimeUnit.SECONDS).blockingSubscribe(PrintLogSubscriber.create());

    }

    /**
     * 创建一个范围发射的被观察者
     */
    private static void range() {
        //从3开始发射，共发射5次
        Observable.range(3, 5).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个被观察者（常用方式）
     */
    private static void just() {
        Observable.just("RxJava Just").subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个周期循环发射的被观察者
     */
    private static void interval() {
        //初始延迟2秒、之后期间延迟1秒
        Observable.interval(2, 1, TimeUnit.SECONDS).blockingSubscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个发射集合的被观察者
     */
    private static void from() {
        String[] arr = new String[]{"1", "2", "3"};
        //数组形式
        Observable.fromArray(arr).subscribe(PrintLogSubscriber.create());

        //集合形式
        Observable.fromIterable(Arrays.asList(arr)).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个只会发射异常的被观察者
     */
    private static void error() {
        Observable.<String>error(new Exception("Throw")).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个永远不会发射的被观察者，即不会调用onNext/onError/onComplete方法
     */
    private static void never() {
        Observable.<String>never().subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个空的被观察者，不会发送onNext，只会触发onComplete
     */
    private static void empty() {
        Observable.<String>empty().subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个延迟订阅的被观察者
     */
    private static void defer() {
        //创建一个延迟订阅的被观察者
        Observable.defer(new Callable<ObservableSource<String>>() {

            @Override
            public ObservableSource<String> call() throws Exception {
                //当调用subscribe时，这里才真正创建被观察者
                return Observable.just("Defer");
            }
        }).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 创建一个被观察者
     */
    private static void create() {
        //使用create操作符创建一个被观察者
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                //发送事件
                emitter.onNext("Hello RxJava!");
                //发送完成事件
                emitter.onComplete();
            }
        });

        //创建一个观察者(订阅者),这里为了简化代码，使用该方式
        Observer<String> observer = PrintLogSubscriber.create();

        //给被观察者设置一个观察者(订阅者)
        observable.subscribe(observer);
    }


}
